package pers.vic.mq.kafka.consumer;

import java.util.HashMap;
import java.util.Map;

import javax.annotation.Resource;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.MessageListener;
import org.springframework.kafka.listener.MessageListenerContainer;
import org.springframework.kafka.listener.config.ContainerProperties;

/**
 * 配置一些KAFKA消费者 测试用 
 * 
 * @author VIC
 *
 */
@Configuration
public class BaseConsumeConfig {

	protected Logger logger = LoggerFactory.getLogger(getClass());

	@Resource(name = "defaultconsumerconfig")
	private Map<String, Object> defaultConsumerFactoryConfigs;
	
	private static volatile int clientId = 1;


	/**
	 * 创建消费者
	 * @param groupId 消费者的组id  一个组中一个主题只会被某个消费者 消费一次
	 * @param topicName  主题id
	 * @param topicPartitions
	 * @param listener
	 * @return
	 */
	protected MessageListenerContainer createConsumer(String groupId, String topicName, Integer topicPartitions,
			MessageListener<String, String> listener) {
		Map<String, Object> consumerFactoryConfigs = new HashMap<>(defaultConsumerFactoryConfigs);
		consumerFactoryConfigs.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
		clientId = clientId +1;
		consumerFactoryConfigs.put("client.id", clientId);
		ConsumerFactory<String, String> consumerFactory = new DefaultKafkaConsumerFactory<>(consumerFactoryConfigs);
		ConcurrentMessageListenerContainer<String, String> container = new ConcurrentMessageListenerContainer<>(
				consumerFactory, new ContainerProperties(topicName));
		container.setConcurrency(topicPartitions);
		container.setupMessageListener(listener);
		return container;
	}

}
